package com.hudson.broker.consumer;


import com.hudson.broker.monitor.BizMonitor;
import com.hudson.broker.monitor.ThreadPoolHelper;
import com.hudson.broker.utils.NettyUtil;
import com.hudson.codec.domain.MessageExt;
import com.hudson.codec.domain.RemotingCommand;
import com.hudson.common.enums.MessageTypeEnum;
import com.hudson.core.util.DateUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @Author: 渔小鲤
 * @DateTime: 2025/7/8 21:15
 **/
public class WaitRequest {
    private static final Logger logger = LoggerFactory.getLogger(WaitRequest.class);

    public static ConcurrentHashMap<String, ConcurrentHashMap<Integer, LinkedBlockingQueue<PullRequest>>> WAIT_REQUESTS = new ConcurrentHashMap<>();

    static {
//        ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(1);
        // 定时清理超时请求（每 1 秒执行一次）
        ThreadPoolHelper.DEFAULT_SCHEDULED_POOL.scheduleAtFixedRateByWatch(() -> {
            for (Map.Entry<String, ConcurrentHashMap<Integer, LinkedBlockingQueue<PullRequest>>> topicEntry : WAIT_REQUESTS.entrySet()) {
                for (Map.Entry<Integer, LinkedBlockingQueue<PullRequest>> queueEntry : topicEntry.getValue().entrySet()) {
                    Iterator<PullRequest> iterator = queueEntry.getValue().iterator();
                    while (iterator.hasNext()) {
                        PullRequest request = iterator.next();
                        if (System.currentTimeMillis() - request.getCreateTime() > 3000) { // 30 秒超时
                            iterator.remove(); // 移除超时请求


                            logger.debug("请求创建时间: {}, 当前时间: {}, 差值:  {} s",
                                    DateUtil.getLocalDateTime(request.getCreateTime()), DateUtil.getLocalDateTime(System.currentTimeMillis()), (System.currentTimeMillis() - request.getCreateTime()) / 1000);

                            logger.debug("返回空对象");

                            logger.debug("准备发送空消息");
                            if (request.getCtx().channel().isActive()) {

                                MessageExt messageExt = MessageExt.builder()
                                        .topic(request.getTopic())
                                        .queueId(request.getQueueId())
                                        .body(null)
                                        .build();
                                NettyUtil.write(request.getCtx().channel(),
                                        RemotingCommand.builder()
                                                .type(MessageTypeEnum.PUSH_MESSAGE_RESP.getType())
                                                .body(MessageExt.list2Bytes(List.of(messageExt)))
                                                .build());
//                                        new Message<>(MessageTypeEnum.PUSH_MESSAGE_RESP.getType(),
//                                        new PushMessageResp(request.getTopic(), request.getQueueId(), 0, ConsumerOffset.getOffset(request.getConsumerGroup(), request.getTopic(), request.getQueueId()), null)));
                                logger.debug("已发送空消息");
                                BizMonitor.CONSUME_SPEED.getAndIncrement();
                            } else {
                                logger.debug("Channel 不活跃，无法发送");
                            }

                        }
                    }
                }
            }
        }, 0, 1, TimeUnit.SECONDS);
    }

    // 不存在数据时，添加到请求队列中
    public static void putPullRequest(String topic, int queueId, PullRequest request) {
        logger.debug("consumer 进入方法");

        ConcurrentHashMap<Integer, LinkedBlockingQueue<PullRequest>> requestMap = WAIT_REQUESTS.computeIfAbsent(topic, k -> new ConcurrentHashMap<>());
        LinkedBlockingQueue<PullRequest> pullRequests = requestMap.computeIfAbsent(queueId, k -> new LinkedBlockingQueue<>());
        pullRequests.offer(request);
    }

    // 当有消息时，从队列中找到请求并返回结果
    public static void notifyMessageArrived(String topic, int queueId) {
//        ConcurrentHashMap<Integer, LinkedBlockingQueue<PullRequest>> requestMap = WAIT_REQUESTS.get(topic);
//        if (requestMap == null) {
//            return;
//        }
//        LinkedBlockingQueue<PullRequest> requestQueue = requestMap.get(queueId);
//        if (requestQueue == null) {
//            return;
//        }
//        PullRequest request;
//        while ((request = requestQueue.poll()) != null) {
//
//            MessageMeta message = CommitLogManger.getMessageByTopic(request.getTopic(), request.getQueueId(), ConsumerOffset.getOffset(request.getConsumerGroup(), request.getTopic(), request.getQueueId()));
//            ConsumerOffset.putOffset(request.getConsumerGroup(), request.getTopic(), request.getQueueId());
//            Integer offset = ConsumerToBrokerHandler.addConsumerOffset(request.getConsumerGroup(), request.getTopic(), request.getQueueId());
        //实际数据
//            NettyUtil.write(request.getCtx().channel(),
//                    new Message<>(MessageTypeEnum.PUSH_MESSAGE_RESP.getType(), new PushMessageResp(request.getTopic(), request.getQueueId(), 1, offset, List.of(message))));
//            NettyUtil.write(request.getCtx().channel(),
//                    RemotingCommandProto.RemotingCommand.newBuilder()
//                            .setType(MessageTypeEnum.PUSH_MESSAGE_RESP.getType())
//                            .setBody(MessageExtProto.MessageExt.newBuilder()
//                                    .setTopic(request.getTopic())
//                                    .setQueueId(request.getQueueId())
//                                    .setBody(ByteString.copyFrom(message.getMessage()))
//                                    .getBody()));
//        BizMonitor.CONSUME_SPEED.getAndIncrement();
//        }


    }
}
